package com.haocjia.esdemo.elastic.config;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

@Configuration
public class ElasticsearchConfig {

    @Value("${es.rest.uris}")
    private String hosts;
    @Value("${es.rest.username}")
    private String username;
    @Value("${es.rest.password}")
    private String password;
    @Value("${es.kas}")
    private Integer kas;

    @Bean
    public RequestOptions requestOptions() {
        return RequestOptions.DEFAULT.toBuilder().build();
    }

    @Bean
    public RestHighLevelClient restHighLevelClient() {

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));


        HttpHost[] httpHosts = Arrays.stream(hosts.split(",")).map(x -> {
            String[] hostInfo = x.split(":");
            return new HttpHost(hostInfo[0], Integer.parseInt(hostInfo[1]));
        }).toArray(HttpHost[]::new);

        RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);

        // 设置连接超时、读取超时、请求超时时间
        restClientBuilder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(30000);
            requestConfigBuilder.setSocketTimeout(60000);
            requestConfigBuilder.setConnectionRequestTimeout(60000);
            return requestConfigBuilder;
        });

        // 处理 Connection reset by peer
        restClientBuilder.setHttpClientConfigCallback(requestConfig ->
                requestConfig.setKeepAliveStrategy((response, context) -> TimeUnit.MINUTES.toMillis(kas)));

        // 设置访问es的用户名密码、线程配置
        restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder
                    .setDefaultCredentialsProvider(credentialsProvider)
                    //设置首次或者超时间停止查询超时问题
                    .setKeepAliveStrategy((response, context) -> Duration.ofMinutes(kas).toMillis())
                    // 配置IO线程
                    .setDefaultIOReactorConfig(IOReactorConfig.custom()
                            .setIoThreadCount(50)
                            .setSoTimeout(60000)
                            .setConnectTimeout(60000)
                            .build());
            return httpClientBuilder;
        });
        return new RestHighLevelClient(restClientBuilder);
    }


}
